package com.imooc.bilibili.config;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.imooc.bilibili.constant.UserMomentsConstant;
import com.imooc.bilibili.entity.UserFollowingInfo;
import com.imooc.bilibili.entity.UserMoments;
import com.imooc.bilibili.service.UserFollowService;
import com.imooc.bilibili.service.impl.WebSocketService;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.core.RedisTemplate;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

/**
 * RocketMQ配置类
 */
@Configuration
public class RocketMqConfig {

    @Value("${rocketmq.name.server.address}")
    private String nameServerAddr;

    @Autowired
    private RedisTemplate<String, String> redisTemplate;

    @Autowired
    private UserFollowService userFollowService;

    @Bean("momentsProducer")
    public DefaultMQProducer getMomentsProducerBean() throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer(UserMomentsConstant.GROUP_MOMENTS);
        //设置 注册中心ip
        producer.setNamesrvAddr(nameServerAddr);
        //启动生产者
        producer.start();
        return producer;
    }

    @Bean("momentsConsumer")
    public DefaultMQPushConsumer getMomentsConsumerBean() throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(UserMomentsConstant.GROUP_MOMENTS);
        consumer.setNamesrvAddr(nameServerAddr);
        //设置订阅的topic
        //*代表当前topic的所有下级主题都订阅
        consumer.subscribe(UserMomentsConstant.TOPIC_MOMENTS, "*");
        //为消费者创建一个监听器，当生产者将消息推送到MQ的时候，消费者通过监听器抓取消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                MessageExt messageExt = list.get(0);
                if (null == messageExt) {
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
                String bodyStr = new String(messageExt.getBody());
                UserMoments userMoments = JSONObject.toJavaObject(JSONObject.parseObject(bodyStr), UserMoments.class);
                Long userId = userMoments.getUserId();
                List<UserFollowingInfo> fanList = userFollowService.getUserFansService(userId);

                //对列表进行遍历，为每个关注用户发送动态
                Iterator<UserFollowingInfo> iterator = fanList.iterator();
                while (iterator.hasNext()) {
                    UserFollowingInfo next = iterator.next();
                    StringBuilder builder = new StringBuilder("subscibed-");
                    builder.append(next.getUserId());
                    String key = builder.toString();
                    builder.setLength(0);

                    String subList = redisTemplate.opsForValue().get(key);
                    List<UserMoments> subscribedList;
                    if (StringUtils.isBlank(subList)) {
                        subscribedList = new ArrayList<>();
                    } else {
                        subscribedList = JSONArray.parseArray(subList, UserMoments.class);
                    }
                    subscribedList.add(userMoments);
                    redisTemplate.opsForValue().set(key, JSONObject.toJSONString(subscribedList));
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }

                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
        return consumer;
    }

    @Bean("barrageProducer")
    public DefaultMQProducer barrageProducer() throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer(UserMomentsConstant.GROUP_BARRAGE);
        //设置 注册中心ip
        producer.setNamesrvAddr(nameServerAddr);
        //启动生产者
        producer.start();
        return producer;
    }

    @Bean("barrageConsumer")
    public DefaultMQPushConsumer barrageConsumer() throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(UserMomentsConstant.TOPIC_BARRAGE);
        consumer.setNamesrvAddr(nameServerAddr);
        //设置订阅的topic
        //*代表当前topic的所有下级主题都订阅
        consumer.subscribe(UserMomentsConstant.TOPIC_BARRAGE, "*");
        //为消费者创建一个监听器，当生产者将消息推送到MQ的时候，消费者通过监听器抓取消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                MessageExt messageExt = list.get(0);
                byte[] msgByte = messageExt.getBody();
                String bodyStr = new String(msgByte);
                JSONObject jsonObject = JSONObject.parseObject(bodyStr);
                String sessionId = jsonObject.getString("sessionId");
                String message = jsonObject.getString("message");
                WebSocketService webSocketService = WebSocketService.WEB_SOCKET_MAP.get(sessionId);
                if (webSocketService.getSession().isOpen()) {
                    try {
                        webSocketService.sendMessage(message);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
                //标记消息被成功消费
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
        return consumer;
    }
}
